/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.util.Properties;

import org.sss.client.Job;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.HashPartitioner2;

public class RankUrlApp {
  private static final int topX = 10;
  public static void usage() {
    System.err.println("Usage:");
    System.err.println("java RankUrlApp gid mapInterval reduceInterval reduceIntervals");
    System.err.println("        gid: input group id (should be with destructive attribute)");
    System.err.println("   interval: interval for whole job tree (in sec)");
    System.err.println("  intervals: how many intervals will be counted in target period");
    System.exit(1);
  }
  /**
   * Usage:
   * java RankUrl gid interval intervals
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);
    String[] others = client.getArgs();
    if (others.length != 3) {
      usage();
    }

    JobEngine engine = new JobEngine(client);

    GroupID logIn = GroupID.createFromString(others[0]);
    GroupID countOut = GroupID.createRandom(engine, false, true);
    engine.getJobBuilder("logmap", ApacheLogMapper.class)
      .addInput(logIn).addOutput(countOut).build();

    GroupID countIn = GroupID.createFromString(countOut.toString());
    GroupID sumOut = GroupID.createRandom(engine, false, true);
    Job.Builder reduceBuilder =
      engine.getJobBuilder("countreduce", ContIntSumReducer.class)
      .addInput(countIn).addOutput(sumOut, Job.partitioner(HashPartitioner2.class));
    Configuration reduceConf = reduceBuilder.getConfiguration();
    reduceConf.setInt("continuous.task.target.intervals", Integer.parseInt(others[2]));
    reduceConf.setBoolean("intsum.filter.top", true);
    reduceConf.setInt("intsum.storage.node",
                      (int) (client.getCluster().getStorageNodes().size() * Math.random()));
    reduceConf.setInt("intsum.storage.slot", sumOut.getNSlot());
    reduceBuilder.build();

    GroupID sumIn = GroupID.createFromString(sumOut.toString());
    GroupID rankOut = GroupID.createRandom(engine, false, true);
    Job.Builder reduceBuilder2 =
      engine.getJobBuilder("rankreduce", ContTopXSelectReducer.class)
      .addInput(sumIn).addOutput(rankOut);
    Configuration reduceConf2 = reduceBuilder2.getConfiguration();
    reduceConf2.setInt("topx.select.rank", topX);
    reduceBuilder2.build();

    engine.setInterval(Long.parseLong(others[1]) * 1000L);
    Properties execInfo = engine.contExec();

    System.out.println("TaskSetID - " + execInfo.getProperty("job.taskset.id"));
    System.out.println("[ApacheLogMapper]");
    System.out.println(" jobID [logmap] - " + execInfo.getProperty("job.logmap.id"));
    System.out.println(" output tuples from logmap will be deployed - " + countOut);

    System.out.println("[ContIntSumReducer]");
    System.out.println(" jobID [countreduce] - " + execInfo.getProperty("job.countreduce.id"));
    System.out.println(" output tuples from countreduce will be deployed - " + sumOut);

    System.out.println("[ContTopXSelectReducer]");
    System.out.println(" jobID [rankreduce] - " + execInfo.getProperty("job.rankreduce.id"));
    System.out.println(" output tuples from rankreduce will be deployed - " + rankOut);

    try {
      while (true) {
        Thread.sleep(1000L);
      }
    }
    catch (InterruptedException e) {
      engine.stopContExec(false);
    }
  }
}
